//////////////////////////////////////////////////////////////////////////////////
//                                                                              //
//  This file is part of the buola project (https://code.google.com/p/buola/).  //
//                                                                              //
//  Copyright(c) 2007-2012 Xavi Gratal                                          //
//  gratal AT gmail DOT com                                                     //
//                                                                              //
//  Buola is free software: you can redistribute it and/or modify               //
//  it under the terms of the GNU General Public License as published by        //
//  the Free Software Foundation, either version 3 of the License, or           //
//  (at your option) any later version.                                         //
//                                                                              //
//  Buola is distributed in the hope that it will be useful,                    //
//  but WITHOUT ANY WARRANTY; without even the implied warranty of              //
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               //
//  GNU General Public License for more details.                                //
//                                                                              //
//  You should have received a copy of the GNU General Public License           //
//  along with buola.  If not, see <http://www.gnu.org/licenses/>.              //
//                                                                              //
//////////////////////////////////////////////////////////////////////////////////

#include <buola/bbox/protocol/csyncconnection.h>

namespace buola { namespace bbox { namespace protocol {

CSyncConnection::CSyncConnection(io::PSocket pSocket,const std::string &pName,io::CFDPoller &pPoller)
    :   CConnectionBase(std::move(pSocket),pName)
    ,   mFDPoller(pPoller)
    ,   mReading(false)
{
    Socket()->SetActivityDelegate(this,io::ENotify::READ);
}

std::unique_ptr<CPacket> CSyncConnection::WaitReply(CPacket &pPacket)
{
    uint32_t lID=pPacket.Header().mSeqNumber;
    
    unique_lock<mutex> lLock(mMutex);

    auto i=mPending.find(lID);
    if(i!=mPending.end()) //it's already there
    {
        auto lPacket=std::move(i->second);
        mPending.erase(i);
        return lPacket;
    }
    
    if(!mReading) //no one is reading, so let's read it ourselves
        return ReadReply(lID,lLock);
    
    auto w=mWaited.emplace(lID,nullptr).first;
    
    while(true)
    {
        mCondVar.wait(lLock);
        
        if(w->second)
        {
            auto lPacket=std::move(w->second);
            mWaited.erase(w);
            return lPacket;
        }
        
        if(!mReading) //we'll read it ourselves
        {
            mWaited.erase(w);
            return ReadReply(lID,lLock);
        }
    }
}

std::unique_ptr<CPacket> CSyncConnection::SendPacketAndWaitReply(CPacket &pPacket)
{
    SendPacket(pPacket);
    return WaitReply(pPacket);
}

std::unique_ptr<CPacket> CSyncConnection::ReadReply(uint32_t pID,unique_lock<mutex> &pLock)
{
    mReading=true;
    pLock.unlock();
    
    while(std::unique_ptr<CPacket> lPacket=ReadPacket(true))
    {
        if(lPacket->Header().Major()==EOpcode::REPLY||lPacket->Header().Major()==EOpcode::ERROR)
        {
            uint32_t lID=lPacket->Header().mSeqReply;
            
            if(lID==pID)
            {
                mReading=false;
                mCondVar.notify_one();
                return lPacket;
            }
            
            pLock.lock();
            
            auto w=mWaited.find(lID);
            if(w!=mWaited.end())
            {
                w->second=std::move(lPacket);
                mCondVar.notify_all();
            }
            else
            {
                mPending.emplace(lID,std::move(lPacket));
            }
            
            pLock.unlock();
        }
        else
        {
            OnPacket(*lPacket);
        }
    }
    
    OnEOF();
    
    return nullptr;
}

void CSyncConnection::OnStreamCanRead(io::CFDStream *pStream)
{
    unique_lock<mutex> lLock(mMutex);

    if(mReading) return;
    
    mReading=true;

    while(std::unique_ptr<CPacket> lPacket=ReadPacket(false))
    {
        if(lPacket->Header().Major()==EOpcode::REPLY||lPacket->Header().Major()==EOpcode::ERROR)
        {
            uint32_t lID=lPacket->Header().mSeqReply;
            auto w=mWaited.find(lID);
            if(w!=mWaited.end())
            {
                w->second=std::move(lPacket);
                mCondVar.notify_all();
            }
            else
            {
                mPending.emplace(lID,std::move(lPacket));
            }
        }
        else
        {
            OnPacket(*lPacket);
        }
    }

    if(AtEOF())
    {
        Socket()->SetActivityDelegate(nullptr,io::ENotify::NONE); //just in case we don't get destroyed as we should
        OnEOF();
    }

    mReading=false;
    mCondVar.notify_one();
}

/*namespace protocol*/ } /*namespace bbox*/ } /*namespace buola*/ }
